# 协程

import asyncio
import time
import random
from typing import List, Coroutine


class AsyncCoordinator:
    """协程协调器，用于管理和同步多个异步任务"""

    def __init__(self):
        self.tasks = []
        self.start_time = None

    async def worker(self, name: str, duration: int):
        """工作协程示例"""
        print(f"[{name}] 开始执行，预计耗时 {duration} 秒")

        # 模拟异步工作
        await asyncio.sleep(duration)

        # 模拟一些异步IO操作
        await self.simulate_io_operation(name)

        print(f"[{name}] 完成！耗时 {duration} 秒")
        return f"{name}-结果"

    async def simulate_io_operation(self, name: str):
        """模拟异步IO操作"""
        io_duration = random.uniform(0.5, 2.0)
        print(f"[{name}] 进行IO操作，耗时 {io_duration:.2f} 秒")
        await asyncio.sleep(io_duration)

    async def data_processor(self, data: str, stage: int):
        """数据处理协程示例"""
        print(f"[处理器-{stage}] 处理数据: {data}")
        process_time = random.uniform(1.0, 3.0)
        await asyncio.sleep(process_time)
        result = f"{data} -> 阶段{stage}完成"
        print(f"[处理器-{stage}] 完成: {result}")
        return result

    async def parallel_execution(self):
        """并行执行多个协程"""
        print("\n" + "=" * 50)
        print("开始并行执行")
        print("=" * 50)

        # 创建多个任务并行执行
        tasks = [
            asyncio.create_task(self.worker(f"任务-{i}", random.randint(2, 5)))
            for i in range(1, 6)
        ]

        # 等待所有任务完成
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print(f"\n并行执行完成，共 {len(results)} 个结果")
        return results

    async def sequential_execution(self):
        """顺序执行协程（演示await的用法）"""
        print("\n" + "=" * 50)
        print("开始顺序执行")
        print("=" * 50)

        results = []

        # 顺序执行
        for i in range(1, 4):
            result = await self.worker(f"顺序任务-{i}", random.randint(1, 3))
            results.append(result)

        print(f"\n顺序执行完成，共 {len(results)} 个结果")
        return results

    async def chained_processing(self):
        """链式处理演示"""
        print("\n" + "=" * 50)
        print("开始链式处理")
        print("=" * 50)

        # 生成初始数据
        initial_data = await self.worker("数据生成器", 1)

        # 链式处理
        stage1 = await self.data_processor(initial_data, 1)
        stage2 = await self.data_processor(stage1, 2)
        stage3 = await self.data_processor(stage2, 3)

        print(f"\n链式处理最终结果: {stage3}")
        return stage3

    async def mixed_execution(self):
        """混合执行模式：部分并行，部分顺序"""
        print("\n" + "=" * 50)
        print("开始混合执行")
        print("=" * 50)

        # 第一阶段：并行执行
        parallel_tasks = [
            self.worker(f"混合并行-{i}", random.randint(1, 3))
            for i in range(1, 4)
        ]
        phase1_results = await asyncio.gather(*parallel_tasks)

        # 第二阶段：顺序处理第一阶段的结果
        final_results = []
        for i, result in enumerate(phase1_results, 1):
            processed = await self.data_processor(result, i)
            final_results.append(processed)

        print(f"\n混合执行完成，共 {len(final_results)} 个最终结果")
        return final_results

    async def run_with_timeout(self):
        """带超时的协程执行"""
        print("\n" + "=" * 50)
        print("开始带超时的执行")
        print("=" * 50)

        try:
            # 设置超时为3秒
            result = await asyncio.wait_for(
                self.worker("超时测试任务", 5),  # 这个任务需要5秒，但超时设为3秒
                timeout=3.0
            )
            print(f"任务完成: {result}")
            return result
        except asyncio.TimeoutError:
            print("任务执行超时！")
            return None

    async def monitor_progress(self):
        """进度监控协程"""
        start = time.time()
        while True:
            elapsed = time.time() - start
            print(f"程序已运行 {elapsed:.1f} 秒")
            await asyncio.sleep(2)  # 每2秒报告一次

    async def run_all_demonstrations(self):
        """运行所有演示"""
        self.start_time = time.time()

        # 启动进度监控（作为后台任务）
        monitor_task = asyncio.create_task(self.monitor_progress())

        try:
            # 执行各种协程模式
            await self.parallel_execution()
            await self.sequential_execution()
            await self.chained_processing()
            await self.mixed_execution()
            await self.run_with_timeout()

        finally:
            # 停止进度监控
            monitor_task.cancel()
            try:
                await monitor_task
            except asyncio.CancelledError:
                print("\n进度监控已停止")

        total_time = time.time() - self.start_time
        print(f"\n所有演示完成！总耗时: {total_time:.2f} 秒")


async def simple_example():
    """简单示例：快速理解协程基础"""
    print("\n" + "=" * 50)
    print("简单示例")
    print("=" * 50)

    async def task(name, seconds):
        print(f"开始 {name}")
        await asyncio.sleep(seconds)
        print(f"完成 {name}")
        return f"{name}-结果"

    # 同时运行多个协程
    results = await asyncio.gather(
        task("快速任务", 1),
        task("中等任务", 2),
        task("慢速任务", 3)
    )

    print(f"简单示例结果: {results}")


async def main():
    """主函数"""
    print("Python协程与事件循环演示")
    print("说明: 本程序演示如何使用asyncio管理异步任务")

    # 运行简单示例
    await simple_example()

    # 运行完整演示
    coordinator = AsyncCoordinator()
    await coordinator.run_all_demonstrations()

    print("\n" + "=" * 50)
    print("程序正常结束，事件循环将关闭")
    print("=" * 50)


def run_program():
    """运行程序的入口函数"""
    try:
        # 获取或创建事件循环
        loop = asyncio.get_event_loop()

        # 设置调试模式（可选）
        loop.set_debug(True)

        print("启动事件循环...")

        # 运行主程序
        loop.run_until_complete(main())

    except KeyboardInterrupt:
        print("\n\n程序被用户中断")
    except Exception as e:
        print(f"\n\n程序执行出错: {e}")
    finally:
        print("🧹 清理资源...")

        # 获取所有运行中的任务并取消
        pending = asyncio.all_tasks(loop)
        for task in pending:
            task.cancel()

        # 等待任务取消完成
        if pending:
            loop.run_until_complete(
                asyncio.gather(*pending, return_exceptions=True)
            )

        # 关闭事件循环
        loop.close()
        print("事件循环已关闭，程序退出")


if __name__ == "__main__":
    run_program()
